﻿using Confluent.Kafka;
using JT808Protocol.Models;
using System;

namespace JT808Gateway.WebHost
{
    public class JT808MsgConsumer
    {
        private readonly KafkaConfig _kafkaConfig;

        private bool _cancelled = false;

        public JT808MsgConsumer(KafkaConfig kafkaConfig)
        {
            _kafkaConfig = kafkaConfig;
        }

        /// <summary>
        /// 订阅收到消息处理事件
        /// </summary>
        /// 参数1：Sim卡号
        /// 参数2：808部标协议包
        public Action<string, JT808Package> EventMsgReceivedHandler { get; set; }

        /// <summary>
        /// 订阅
        /// </summary>
        public void Subscribe()
        {
            if (string.IsNullOrEmpty(_kafkaConfig.JT808GwRecevCmdTopic))
                return;

            var consumerBuilder = new ConsumerBuilder<string, JT808Package>(new ConsumerConfig
            {
                BootstrapServers = $"{_kafkaConfig.Host}:{_kafkaConfig.Port}",
                AutoOffsetReset = AutoOffsetReset.Earliest
            });

            using (var consumer = consumerBuilder.Build())
            {
                consumer.Subscribe(_kafkaConfig.JT808GwRecevCmdTopic);

                while (!_cancelled)
                {
                    var cr = consumer.Consume();
                    if (EventMsgReceivedHandler != null) //抛出到回调
                        EventMsgReceivedHandler(cr.Message.Key, cr.Message.Value);
                }

                consumer.Close();
            }
        }
    }
}